package it.antonio.ilpizze.db.core;

import it.antonio.ilpizze.db.Id;
import it.antonio.ilpizze.db.Record;
import it.antonio.ilpizze.db.core.id.FilePositionId;
import it.antonio.ilpizze.db.core.id.MemoryTableId;
import it.antonio.ilpizze.db.core.id.RecordFilePositionId;
import it.antonio.ilpizze.db.io.IOConstants;
import it.antonio.ilpizze.db.io.IOFile;
import it.antonio.ilpizze.func.ComputeIterator;
import it.antonio.ilpizze.func.Pair;
import it.antonio.ilpizze.func.Predicate;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

public class RecordManager implements RecordRecoverManager, IOConstants {
	
	private IOFile file;
	private TransactionManager txManager;
	
	private static final byte FREE = 1; 
	private static final byte USED = 2;
	
	private static final byte FIRST_RECORD_POSITION = SIZE_LONG;
	private static final byte FILE_SIZE_POSITION = 0;
	
	public RecordManager(IOFile file, TransactionManager txManager) {
		this.file = file;
		if(this.file.size() == 0){
			this.file.writeLong(FIRST_RECORD_POSITION);
		}
		this.txManager = txManager;
	}

	public FilePositionId allocate(Record record){
		file.position(FILE_SIZE_POSITION);
		final long position = file.readLong();
		file.position(FILE_SIZE_POSITION);
		file.writeLong(position + SIZE_BYTE + SIZE_INT + record.size());
		return RecordFilePositionId.atPosition(position);
	}
	
	
	public void allocateForMemoryTable(MemoryTableId id, Record record) {
		file.position(FILE_SIZE_POSITION);
		final long position = file.readLong();
		file.position(FILE_SIZE_POSITION);
		file.writeLong(position + SIZE_BYTE + SIZE_INT + record.size());
		id.persisted(position);
	}
	
	public void save(FilePositionId id, Record record) {
		final long position = id.pos();
		file.position(position);

		if (txManager.autoCommit() == false) {
			txManager.add(new TransactionUnit() {

				@Override
				public void rollback() {
					file.position(position);
					file.writeByte(FREE);
				}

				@Override
				public void commit() {
				}

				@Override
				public void log(IOFile txLog) {
					txLog.writeByte(SAVE_RECOVER);
					txLog.writeLong(position);
				}
			});

		}

		file.writeByte(USED);
		file.writeInt(record.size());

		for (Pair<String, Object> p : record) {
			byte[] keyBytes = p.first().getBytes();

			file.writeInt(keyBytes.length);
			file.writeBytes(keyBytes);

			Object value = p.second();

			Types type = Types.decode(value);
			file.writeByte(type.typeByte());
			type.write(file, value);

		}
	}
	
	public Record read(FilePositionId id) {
		file.position(id.pos());

		byte used = file.readByte();
		if (used == FREE) {
			return null;
		}

		int size = file.readInt();

		return readRecord(size);
	}
	
	public Record remove(FilePositionId id) {
		final long position = id.pos();
		file.position(position);

		byte used = file.readByte();

		if (used == USED) {
			if (txManager.autoCommit() == false) {
				txManager.add(new TransactionUnit() {

					@Override
					public void rollback() {
						file.position(position);
						file.writeByte(USED);
					}

					@Override
					public void commit() {
					}

					@Override
					public void log(IOFile txLog) {
						txLog.writeByte(REMOVE_RECOVER);
						txLog.writeLong(position);
					}

				});
			}

			file.position(position);
			file.writeByte(FREE);
			int size = file.readInt();

			return readRecord(size);
		}
		return null;
		
	}
	
	public Iterator<Pair<Id, Record>> load(Predicate<Record> pred) {
		return new RecordIterator(file, pred);
	}
	
	private Record readRecord(int size) {
		Record record = new Record();
		
		while (record.size() < size) {
			int keySize = file.readInt();
			
			byte[] b = file.readBytes(keySize);
			String key = new String(b);
			
			byte typeByte = file.readByte();
			Types type = Types.decode(typeByte);
			
			Object value = type.read(file);
			
			record.put(key, value);
			
		}
		return record;
	}
	
	
	
	private class RecordIterator extends ComputeIterator<Pair<Id,Record>>  {
		
		private IOFile file;
		private Predicate<Record> pred;
		private long fileSize;
		
		private RecordIterator(IOFile file, Predicate<Record> pred) {
			this.file = file;
			this.pred = pred;
			this.file.position(FILE_SIZE_POSITION);
			this.fileSize = this.file.readLong();
		}
		
		protected Pair<Id,Record> computeNext() {
			while (true) {
				long position = file.position();
				Id id = RecordFilePositionId.atPosition(position);
				
				if(position >= fileSize){
					return finished();
				}
				
				byte used = file.readByte();
				int size = file.readInt();
				
				if(used == FREE){
					file.position(position + SIZE_BYTE + SIZE_INT + size);
					continue;
				}
				
				Record record = readRecord(size);
				
				if(pred.match(record)){
					return Pair.from(id, record);
				} else {
					continue;
				}
				
			}
		}
	}



	@Override
	public void recoverRemove(long position) {
		file.position(position);
		file.writeByte(USED);
	}

	@Override
	public void recoverSave(long position) {
		file.position(position);
		file.writeByte(FREE);
		
	}

	public void runMaintenance(DbFileSystem dbFs) {
		Iterator<Pair<Id, Record>> it = load(new Predicate<Record>() {

			@Override
			public boolean match(Record elem) {
				return true;
			}
		});
		List<Pair<Id,Record>> records = new LinkedList<Pair<Id,Record>>();
		while(it.hasNext()) records.add(it.next());
		
		file.close();
		while(file.delete());
		file = dbFs.openRecordFile();
		this.file.writeLong(FIRST_RECORD_POSITION);
		
		for(Pair<Id, Record> record: records){
			// allocate
			file.position(FILE_SIZE_POSITION);
			long position = file.readLong();
			file.position(FILE_SIZE_POSITION);
			file.writeLong(position + SIZE_BYTE + SIZE_INT + record.second().size());
			
			
			// save
			file.position(position);
			file.writeByte(USED);
			file.writeInt(record.second().size());
			
			for(Pair<String,Object> p: record.second()){
				byte[] keyBytes = p.first().getBytes();
				
				file.writeInt(keyBytes.length);
				file.writeBytes(keyBytes);

				Object value = p.second();
				
				Types type = Types.decode(value);
				file.writeByte(type.typeByte());
				type.write(file, value);
				
			}
			
		}
		
	}

	

}
